package com.tee.rocketmqstudy.demo;

import com.alibaba.fastjson.JSON;
import com.tee.rocketmqstudy.constant.MqConstant;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.junit.Test;

import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * @author tee
 * @date 2025/9/8
 */
public class ASimpleTest {

    @Test
    public void simpleProduce() throws Exception {
        // 创建一个生产者
        DefaultMQProducer producer = new DefaultMQProducer("test-producer-group");

        // 连接nameserver
        producer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR);

        producer.start();

        Message message = new Message("order-fallback","questionId:1234".getBytes());

        //发送消息
        SendResult sendResult = producer.send(message);
        System.out.println(sendResult);
        producer.shutdown();

    }

    @Test
    public void simpleConsume() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("test-producer-group");

        consumer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR);
        // *表示订阅这个主题中的所有消息
        consumer.subscribe("order-fallback","*");

        //注册一个监听器
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                // 这个就是消费的方法
                System.out.println("我是消费者");
                System.out.println(list.get(0));
                System.out.println("========================>");
                System.out.println(new String(list.get(0).getBody()));
                System.out.println("消息消费上下文" + JSON.toJSONString(consumeConcurrentlyContext));
                // 返回值 很关键CONSUME_SUCCESS //表示成功,直接出队 // RECONSUME_LATER表示失败,会重新回到队列
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        consumer.start();
        CountDownLatch countDownLatch = new CountDownLatch(1);
        countDownLatch.await();

    }
}
